package redis.inaction.ratelimiter.filter.ratelimit;

import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

@Slf4j
public class LeakBucketRateLimiter {
    private static final String QUEUE_ITEM = "x";
    private final double leakRate;
    private long lastLeakTime;
    private final ArrayBlockingQueue<String> queue;
    private final Executor executor = Executors.newSingleThreadExecutor();
    private final Lock lock = new ReentrantLock();
    private volatile boolean leaking = true;

    public LeakBucketRateLimiter(int capacity, double leakRate) {
        this.leakRate = leakRate;
        this.lastLeakTime = System.currentTimeMillis();
        queue = new ArrayBlockingQueue<>(capacity);
        startLeakTask();
    }

    /**
     * 表示一个请求进来进行入桶操作，桶满则被限流
     *
     * @return
     */
    public boolean isAllow() {
        leakWater();
        return queue.offer(QUEUE_ITEM);
    }

    private void startLeakTask() {
        executor.execute(this::leakWater);
    }

    private void leakWater() {
        while (leaking) {
            if (lock.tryLock()) {
                try {
                    long now = System.currentTimeMillis();
                    int leakNum = (int) ((now - lastLeakTime) * leakRate);
                    for (int i = 0; i < leakNum; i++) {
                        if (queue.poll() == null) {
                            break;
                        }
                    }
                    lastLeakTime = now;
                } finally {
                    lock.unlock();
                }
            }
            sleepOneSecond();
        }
    }

    private void sleepOneSecond(){
        try {
            TimeUnit.SECONDS.sleep(1);
        } catch (InterruptedException e) {
            log.warn("Interrupted");
        }
    }
}
